package com.surfilter.massdata.spark.input.ipcheck;

import java.io.File;
import java.io.IOException;
import java.net.URI;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;

import com.act.sparkanalyz.input.ISparkInput;
import com.act.sparkanalyz.service.impl.SparkService.ServiceContext;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.surfilter.massdata.spark.bean.ipcheck.ActiveIp;
import com.surfilter.massdata.spark.bean.ipcheck.IpBean;
import com.surfilter.massdata.spark.util.ipcheck.CommonUtil;
import com.surfilter.massdata.spark.util.ipcheck.ConvertUtil;
import com.surfilter.massdata.spark.util.ipcheck.IpCheckConstants;
import com.surfilter.massdata.spark.util.ipcheck.IpCheckUtil;
import com.surfilter.massdata.spark.util.ipcheck.IpConvertUtil;
import com.surfilter.massdata.spark.util.ipcheck.IpUtils;

import scala.Tuple2;

/** 
* @author  hutao
* @date 创建时间：2017年6月1日 下午3:22:24 
* @Description：活跃IP输入数据
* @version 1.0 
* 
**/
public class IpCheckActiveIpInput implements ISparkInput {
	private static final Log log = LogFactory.getLog(IpCheckActiveIpInput.class);
	private static final long serialVersionUID = 1L;
	private String filePath;
	private String activceTableName;
	@Override
	public Map<String, DataFrame> read(ServiceContext serviceContext, Map<String, String> commandMap) throws Exception {
		// TODO Auto-generated method stub
		Map<String, DataFrame> result = new HashMap<String, DataFrame>();
		log.info("==============IpCheckActiveIpInput begin====================");
		if(null!=filePath){
			if(filePath.endsWith(File.separator)){
				filePath=filePath.substring(0, filePath.length()-1);
			}
		}
		if(commandMap.containsKey("-d")){
			String date= commandMap.get("-d");
			if(null!=date && date.length()==8){
				filePath+="/"+date.substring(0,4)+"/"+date.substring(4,6)+"/"+date.substring(6,8)+"";
				if(existsPath(filePath)){
					final Broadcast<Map<String, Long>> broadcastBr3010Map=serviceContext.getCtx().broadcast(IpCheckUtil.getBr3010Map());
					System.err.println("activeIpFilePath>>>>>>>>>>>>>>>>>>>>"+filePath);
					JavaRDD<IpBean> activeIpRdd=serviceContext.getCtx().textFile(filePath).mapPartitions(new FlatMapFunction<Iterator<String>, Tuple2<Long, ActiveIp>>() {
						private static final long serialVersionUID = 1L;
						ObjectMapper mapper = new ObjectMapper();
						ObjectWriter writer=mapper.writer();
						List<Tuple2<Long, ActiveIp>> list=new ArrayList<Tuple2<Long, ActiveIp>>();
						Map<String, Long> br3010Map=broadcastBr3010Map.getValue();
						@Override
						public Iterable<Tuple2<Long, ActiveIp>> call(Iterator<String> t) throws Exception {
							while(t.hasNext()){
								String line = t.next();
								String[] arr = line.split("\t");
								if(arr.length == 7){
									String ip = arr[0];
									String bh = arr[1];
									//判断是不是ip
									if(!CommonUtil.isEmpty(ip) && IpUtils.isValidIP(ip)){
										//有一个字段为空就过滤掉
										Long bbdwId = br3010Map.get(bh);
										if(null!=bbdwId&&!CommonUtil.isEmpty(bh)){
											String houseCode = arr[2];
											String lastTime = arr[5];
											ActiveIp activeIp = new ActiveIp();
											activeIp.setBbdwId(bbdwId);
											Long longIp=IpUtils.ipToLong(ip);
											activeIp.setIpLong(longIp);
											activeIp.setQsip(longIp);
											activeIp.setZzip(longIp);
											activeIp.setHouseCode(houseCode);
											activeIp.setKeyTime(Long.parseLong(lastTime));
											activeIp.setFindTime(System.currentTimeMillis());
											activeIp.setStorageType(IpCheckConstants.ACTIVEIP);
											String json = writer.writeValueAsString(activeIp);
											activeIp.setJson(json);
											list.add(new Tuple2<Long, ActiveIp>(longIp,activeIp));
										}
									}
									
								}
							}
							return list;
						}
					}).mapToPair(new PairFunction<Tuple2<Long,ActiveIp>, Long, ActiveIp>() {
						private static final long serialVersionUID = 1L;
						
						@Override
						public Tuple2<Long, ActiveIp> call(Tuple2<Long, ActiveIp> t) throws Exception {
							return new Tuple2<Long,ActiveIp>(t._1,t._2);
						}
						
					}).groupByKey().map(new Function<Tuple2<Long,Iterable<ActiveIp>>, IpBean>() {
						
						private static final long serialVersionUID = 1L;
						
						@Override
						public IpBean call(Tuple2<Long, Iterable<ActiveIp>> iter) throws Exception {
							
							ActiveIp currentActiveIp = new ActiveIp();
							currentActiveIp.setKeyTime(-1l);
							for(ActiveIp activeIp : iter._2){
								ActiveIp newIp = IpConvertUtil.copyActiveIp(activeIp);
								if(newIp.getKeyTime() > currentActiveIp.getKeyTime()){
									currentActiveIp.setBbdwId(newIp.getBbdwId());
									currentActiveIp.setHouseCode(newIp.getHouseCode());
									currentActiveIp.setIpLong(newIp.getIpLong());
									currentActiveIp.setKeyTime(newIp.getKeyTime());
									currentActiveIp.setJson(newIp.getJson());;
								}
							}
							IpBean ipBean = ConvertUtil.convertActiveIp2IpBean(new IpBean(),currentActiveIp);
							
							return ipBean;
						}
					});
					
					System.out.println("activeIPCount>>>>>>>>>>>>>>>>>>"+activeIpRdd.count());
					DataFrame activeIpDF = serviceContext.getSqlCtx().createDataFrame(activeIpRdd, IpBean.class);
					result.put(activceTableName, activeIpDF);
				}
			}
		}
		
		log.info("==============IpCheckActiveIpInput end====================");
		return result;
	}
	public static boolean existsPath(String dst) throws IOException{
		boolean flag=false;
		Configuration conf = new Configuration();
		FileSystem fs = null;
		try {
			fs = FileSystem.get(URI.create(dst), conf);
			flag=fs.exists(new Path(dst));
		} catch (IOException e) {
		} finally {
			if (fs != null)
				fs.close();
		}
		return flag;

	}

}
